/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.connector.jdbc;

import org.apache.flink.annotation.Experimental;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.io.DefaultInputSplitAssigner;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.io.RichInputFormat;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.internal.connection.JdbcConnectionProvider;
import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider;
import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider;
import org.apache.flink.core.io.GenericInputSplit;
import org.apache.flink.core.io.InputSplit;
import org.apache.flink.core.io.InputSplitAssigner;
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Array;
import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.Arrays;

/**
 * InputFormat to read data from a database and generate Rows. 
 * The InputFormat has to be configured using the supplied InputFormatBuilder. 
 * A valid RowTypeInfo must be properly configured in the builder, e.g.:
 * 
 * 用于从数据库读取数据并生成行。必须使用提供的 InputFormatBuilder配置InputFormat。 
 * 必须在构建器中正确配置有效的 RowTypeInfo，
 * 
 * <pre><code>
 * TypeInformation<?>[] fieldTypes = new TypeInformation<?>[] {
 * 	BasicTypeInfo.INT_TYPE_INFO,
 * 	BasicTypeInfo.STRING_TYPE_INFO,
 * 	BasicTypeInfo.STRING_TYPE_INFO,
 * 	BasicTypeInfo.DOUBLE_TYPE_INFO,
 * 	BasicTypeInfo.INT_TYPE_INFO
 * };
 *
 * RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes);
 *
 * JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat()
 * 			.setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
 * 			.setDBUrl("jdbc:derby:memory:ebookshop")
 * 			.setQuery("select * from books")
 * 			.setRowTypeInfo(rowTypeInfo)
 * 			.finish();
 * </code></pre>
 *
 * <p>In order to query the JDBC source in parallel, you need to provide a parameterized query
 * template (i.e. a valid {@link PreparedStatement}) and a {@link JdbcParameterValuesProvider} 
 * which provides binding values for the query parameters. E.g.:
 * 
 * 为了并行查询 JDBC 源，您需要提供一个参数化查询模板（即有效的 {@linkPreparedStatement}）
 * 和一个为查询参数提供绑定值的 {@linkJdbcParameterValuesProvider}。 
 * 
 * <pre><code>
 *
 * Serializable[][] queryParameters = new String[2][1];
 * queryParameters[0] = new String[]{"Kumar"};
 * queryParameters[1] = new String[]{"Tan Ah Teck"};
 *
 * JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat()
 * 			.setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
 * 			.setDBUrl("jdbc:derby:memory:ebookshop")
 * 			.setQuery("select * from books WHERE author = ?")
 * 			.setRowTypeInfo(rowTypeInfo)
 * 			.setParametersProvider(new JdbcGenericParameterValuesProvider(queryParameters))
 * 			.finish();
 * </code></pre>
 *
 * @see Row
 * @see JdbcParameterValuesProvider
 * @see PreparedStatement
 * @see DriverManager
 */
// @Experimental是一个注解，通常用于标记实验性质的 API或功能。
// 它的作用是提醒开发者该 API或功能可能会在未来的版本中发生变化，不建议在生产环境中使用。
@Experimental
public class JdbcInputFormat extends RichInputFormat<Row, InputSplit>
        implements ResultTypeQueryable<Row> {

    protected static final long serialVersionUID = 2L;

    protected static final Logger LOG = LoggerFactory.getLogger(JdbcInputFormat.class);

    // 配置 JDBC输入格式，例如如何获取 JDBC连接、如何构建查询语句、结果集的类型和并发性以及行类型信息等
    // connectionProvider：JDBC连接提供程序，用于提供JDBC连接。
    protected JdbcConnectionProvider connectionProvider;

    // 查询模板，用于构建查询语句。
    protected String queryTemplate;

    // 结果集类型，用于指定结果集的类型。
    protected int resultSetType;

    // 结果集并发性，用于指定结果集的并发性。
    protected int resultSetConcurrency;

    // 行类型信息，用于描述查询结果集中的每一行数据的类型信息。
    protected RowTypeInfo rowTypeInfo;

    // 预编译的 SQL语句，用于执行查询操作。
    protected transient PreparedStatement statement;

    // 查询结果集。
    protected transient ResultSet resultSet;

    // 每次从数据库中获取的记录数。
    protected int fetchSize;
    
    // Boolean to distinguish between default value and explicitly set autoCommit mode.
    // 用于区分默认值和显式设置 autoCommit 模式的布尔值。
    protected Boolean autoCommit;

    // hasNext用于判断是否还有下一条记录
    protected boolean hasNext;

    // parameterValues用于存储查询参数值
    protected Object[][] parameterValues;

    public JdbcInputFormat() {}

    // 获取查询结果集的行类型信息。
    // 它返回一个 RowTypeInfo对象，该对象描述了查询结果集中每一行数据的类型信息，
    // 包括每一列的名称和类型。
    @Override
    public RowTypeInfo getProducedType() {
        return rowTypeInfo;
    }

    @Override
    public void configure(Configuration parameters) {
        // do nothing here
    }

    // 用于读取 JDBC数据源中的数据，并将其转换为 Flink数据集。
    // 这段代码是 JdbcInputFormat类的 openInputFormat方法的实现。
    // 该方法在JdbcInputFormat对象被打开时被调用，
    // 用于初始化JDBC连接和预编译的SQL语句。
    @Override
    public void openInputFormat() {
        // called once per inputFormat (on open)
        try {
            // 获取JDBC连接，
            Connection dbConn = connectionProvider.getOrEstablishConnection();

            // 并根据需要设置自动提交选项。
            // set autoCommit mode only if it was explicitly configured.
            // keep connection default otherwise.
            if (autoCommit != null) {
                dbConn.setAutoCommit(autoCommit);
            }
            // 使用预编译的 SQL语句创建 PreparedStatement对象，
            // PreparedStatement是一个预编译的 SQL语句对象，可以执行带有参数的 SQL查询，提高查询效率。
            // 它是 Statement接口的子接口，具有比 Statement更多的功能和优势。
            statement = dbConn.prepareStatement(queryTemplate, resultSetType, resultSetConcurrency);
            
            // 并根据需要设置获取记录数。如果 fetchSize的值为 Integer.MIN_VALUE或大于 0，
            if (fetchSize == Integer.MIN_VALUE || fetchSize > 0) {
                // 则设置PreparedStatement对象的获取记录数为 fetchSize。
                statement.setFetchSize(fetchSize);
            }
        } catch (SQLException se) {
            // 在获取 JDBC连接或创建 PreparedStatement对象时发生异常，
            // 则抛出 IllegalArgumentException异常。
            throw new IllegalArgumentException("open() failed." + se.getMessage(), se);
        } catch (ClassNotFoundException cnfe) {
            throw new IllegalArgumentException(
                    "JDBC-Class not found. - " + cnfe.getMessage(), cnfe);
        }
    }

    //该方法在 JdbcInputFormat对象被关闭时被调用，用于释放JDBC连接和资源。
    @Override
    public void closeInputFormat() {
        // called once per inputFormat (on close)
        try {
            if (statement != null) {
                statement.close();
            }
        } catch (SQLException se) {
            LOG.info("Inputformat Statement couldn't be closed - " + se.getMessage());
        } finally {
            statement = null;
        }

        connectionProvider.closeConnection();

        parameterValues = null;
    }

    /**
     * Connects to the source database and executes the query in a <b>parallel fashion</b> if this
     * {@link InputFormat} is built using a parameterized query (i.e. using a {@link
     * PreparedStatement}) and a proper {@link JdbcParameterValuesProvider}, in a <b>non-parallel
     * fashion</b> otherwise.
     * 
     * 如果此 {@link InputFormat}是使用参数化查询（即使用 {@linkPreparedStatement}）
     * 和适当的 {@linkJdbcParameterValuesProvider 构建的，
     * 则连接到源数据库并以<b>并行方式</b>执行查询 }，
     * 否则以<b>非并行方式</b>。
     * 
     * 
     * @param inputSplit which is ignored if this InputFormat is executed as a non-parallel source,
     *     a "hook" to the query parameters otherwise (using its <i>splitNumber</i>)
     *  如果此InputFormat作为非并行源执行，则会被忽略，否则是查询参数的 “挂钩”（使用其 <i>splitNumber</i>）
     *
     * @throws IOException if there's an error during the execution of the query
     */
    
    //该方法在打开JdbcInputFormat对象时被调用，用于执行查询操作并获取结果集。
    @Override
    public void open(InputSplit inputSplit) throws IOException {
        try {
            // 检查输入参数inputSplit和parameterValues是否为空。
            if (inputSplit != null && parameterValues != null) {
                // 根据输入参数的索引值，依次获取查询参数的值，
                for (int i = 0; i < parameterValues[inputSplit.getSplitNumber()].length; i++) {
                   
                    // 用于获取查询参数的值的，它是一个二维数组parameterValues中的元素
                    Object param = parameterValues[inputSplit.getSplitNumber()][i];

                    // 并根据参数的类型设置PreparedStatement对象的参数值。
                    // 支持的参数类型包括String、Long、Integer、Double、Boolean、Float、BigDecimal、
                    // Byte、Short、Date、Time、Timestamp和Array。
                    if (param instanceof String) {
                        statement.setString(i + 1, (String) param);
                    } else if (param instanceof Long) {
                        statement.setLong(i + 1, (Long) param);
                    } else if (param instanceof Integer) {
                        statement.setInt(i + 1, (Integer) param);
                    } else if (param instanceof Double) {
                        statement.setDouble(i + 1, (Double) param);
                    } else if (param instanceof Boolean) {
                        statement.setBoolean(i + 1, (Boolean) param);
                    } else if (param instanceof Float) {
                        statement.setFloat(i + 1, (Float) param);
                    } else if (param instanceof BigDecimal) {
                        statement.setBigDecimal(i + 1, (BigDecimal) param);
                    } else if (param instanceof Byte) {
                        statement.setByte(i + 1, (Byte) param);
                    } else if (param instanceof Short) {
                        statement.setShort(i + 1, (Short) param);
                    } else if (param instanceof Date) {
                        statement.setDate(i + 1, (Date) param);
                    } else if (param instanceof Time) {
                        statement.setTime(i + 1, (Time) param);
                    } else if (param instanceof Timestamp) {
                        statement.setTimestamp(i + 1, (Timestamp) param);
                    } else if (param instanceof Array) {
                        statement.setArray(i + 1, (Array) param);
                    } else {
                        // extends with other types if needed
                        // 如果参数类型不在支持的范围内，则抛出IllegalArgumentException异常。
                        // extends with other types if needed
                        throw new IllegalArgumentException(
                                "open() failed. Parameter "
                                        + i
                                        + " of type "
                                        + param.getClass()
                                        + " is not handled (yet).");
                    }
                }
                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            String.format(
                                    "Executing '%s' with parameters %s",
                                    queryTemplate,
                                    Arrays.deepToString(
                                            parameterValues[inputSplit.getSplitNumber()])));
                }
            }
            // 执行查询操作并获取结果集，
            resultSet = statement.executeQuery();
            // 并通过调用 resultSet.next()方法判断是否还有下一条记录。
            hasNext = resultSet.next();
        } catch (SQLException se) {
            
            // 如果在执行查询操作时发生异常，则抛出 IllegalArgumentException异常。
            throw new IllegalArgumentException("open() failed." + se.getMessage(), se);
        }
    }

    /**
     * Closes all resources used.  关闭所有使用的资源。
     *
     * @throws IOException Indicates that a resource could not be closed.
     *                     指示无法关闭资源。
     */
    @Override
    public void close() throws IOException {
        if (resultSet == null) {
            return;
        }
        try {
            resultSet.close();
        } catch (SQLException se) {
            LOG.info("Inputformat ResultSet couldn't be closed - " + se.getMessage());
        }
    }

    /**
     * Checks whether all data has been read.
     * 检查是否所有数据都已读取。
     *
     * @return boolean value indication whether all data has been read.
     *         布尔值指示是否所有数据都已读取。
     * @throws IOException
     */
    @Override
    public boolean reachedEnd() throws IOException {
        return !hasNext;
    }

    /**
     * Stores the next resultSet row in a tuple.
     * 该方法用于获取结果集中的下一条记录，并将其转换为Flink数据集中的Row对象。
     * @param reuse row to be reused.  要重用的行。
     * @return row containing next {@link Row}
     * @throws IOException
     */
    @Override
    public Row nextRecord(Row reuse) throws IOException {
        try {
            if (!hasNext) {
                return null;
            }
            // "reuse.setField"是 Flink中 Tuple和 Row类的方法，用于设置 Tuple和 Row对象中某个字段的值。
            // 在这段代码中，通过循环遍历 Row对象的每个字段，依次从结果集中读取对应的字段值，
            // 并通过调用 "reuse.setField"方法将其设置为 Row对象的对应字段值。
            // 具体而言，"reuse"是一个Row对象，"pos"是字段的索引，
            // "resultSet.getObject(pos + 1)"是结果集中对应字段的值。
            for (int pos = 0; pos < reuse.getArity(); pos++) {
                reuse.setField(pos, resultSet.getObject(pos + 1));
            }
            // update hasNext after we've read the record
            hasNext = resultSet.next();
            return reuse;
        } catch (SQLException se) {
            throw new IOException("Couldn't read data - " + se.getMessage(), se);
        } catch (NullPointerException npe) {
            throw new IOException("Couldn't access resultSet", npe);
        }
    }

    // 获取输入格式的统计信息。
    // 在方法中，它接受一个缓存的统计信息参数cachedStatistics，并直接返回该缓存的统计信息。
    @Override
    public BaseStatistics getStatistics(BaseStatistics cachedStatistics) throws IOException {
        return cachedStatistics;
    }

    // 该方法用于创建输入分片，以便将数据源分成多个部分并行读取。
    @Override
    public InputSplit[] createInputSplits(int minNumSplits) throws IOException {
        
        // 首先检查查询参数是否为空。
        if (parameterValues == null) {
            
            // 如果为空，则返回一个包含单个分片的GenericInputSplit数组。
            return new GenericInputSplit[] {new GenericInputSplit(0, 1)};
        }
        
        // 否则，根据查询参数的数量创建对应数量的分片，并将它们存储在GenericInputSplit数组中。
        // 每个分片都包含一个唯一的分片编号和总分片数。
        GenericInputSplit[] ret = new GenericInputSplit[parameterValues.length];
        
        // 这段代码是一个循环，它用于创建一个包含多个 GenericInputSplit 对象的数组 ret。
        // 循环的目的是为数组中的每个元素都赋值一个新的 GenericInputSplit 对象，
        // 其中 i 是索引，ret.length 是数组的长度。
        for (int i = 0; i < ret.length; i++) {
            ret[i] = new GenericInputSplit(i, ret.length);
        }
        
        //最后，该方法返回填充好的数组ret。
        return ret;
    }

    /**
     * getInputSplitAssigner方法接受一个 InputSplit类型的数组 inputSplits，
     * 并返回一个 InputSplitAssigner 对象，用于指定输入分片的分配方式。
     *
     * @param inputSplits
     * @return
     */
    @Override
    public InputSplitAssigner getInputSplitAssigner(InputSplit[] inputSplits) {
        
        // 在这里，该方法返回一个 DefaultInputSplitAssigner 对象，它将输入分片平均分配给可用的任务插槽。
        return new DefaultInputSplitAssigner(inputSplits);
    }

    @VisibleForTesting
    protected PreparedStatement getStatement() {
        return statement;
    }

    @VisibleForTesting
    protected Connection getDbConn() {
        return connectionProvider.getConnection();
    }

    /**
     * A builder used to set parameters to the output format's configuration in a fluent way.
     * 用于以流畅的方式将参数设置为输出格式配置的构建器。
     * @return builder
     */
    public static JdbcInputFormatBuilder buildJdbcInputFormat() {
        return new JdbcInputFormatBuilder();
    }

    /** Builder for {@link JdbcInputFormat}. */
    //该类用于构建JdbcInputFormat对象，并设置JDBC数据源的相关参数。
    public static class JdbcInputFormatBuilder {
        // 定义JdbcInputFormatBuilder类的两个成员变量：
        // 1)JdbcConnectionOptionsBuilder 对象 connOptionsBuilder
        // 2)JdbcInputFormat对象 format。

        // connOptionsBuilder用于构建JDBC连接选项，
        private final JdbcConnectionOptions.JdbcConnectionOptionsBuilder connOptionsBuilder;

        // format用于设置 JDBC输入格式的相关参数。
        private final JdbcInputFormat format;

        /**
         * 这段代码实现了一个JdbcInputFormatBuilder类，用于构建JdbcInputFormat对象。
         */
        public JdbcInputFormatBuilder() {
            // 在类中，首先创建一个JdbcConnectionOptionsBuilder对象和一个JdbcInputFormat对象。
            this.connOptionsBuilder = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder();
            this.format = new JdbcInputFormat();
            // using TYPE_FORWARD_ONLY for high performance reads
            this.format.resultSetType = ResultSet.TYPE_FORWARD_ONLY;
            this.format.resultSetConcurrency = ResultSet.CONCUR_READ_ONLY;
        }

        // 然后，通过一系列set方法设置JDBC数据源的相关参数，
        // 例如用户名、密码、驱动程序名称、数据库URL、查询语句、结果集类型、结果集并发性等。
        public JdbcInputFormatBuilder setUsername(String username) {
            connOptionsBuilder.withUsername(username);
            return this;
        }

        public JdbcInputFormatBuilder setPassword(String password) {
            connOptionsBuilder.withPassword(password);
            return this;
        }

        public JdbcInputFormatBuilder setDrivername(String drivername) {
            connOptionsBuilder.withDriverName(drivername);
            return this;
        }

        public JdbcInputFormatBuilder setDBUrl(String dbURL) {
            connOptionsBuilder.withUrl(dbURL);
            return this;
        }

        public JdbcInputFormatBuilder setQuery(String query) {
            format.queryTemplate = query;
            return this;
        }

        public JdbcInputFormatBuilder setResultSetType(int resultSetType) {
            format.resultSetType = resultSetType;
            return this;
        }

        public JdbcInputFormatBuilder setResultSetConcurrency(int resultSetConcurrency) {
            format.resultSetConcurrency = resultSetConcurrency;
            return this;
        }

        public JdbcInputFormatBuilder setParametersProvider(
                JdbcParameterValuesProvider parameterValuesProvider) {
            format.parameterValues = parameterValuesProvider.getParameterValues();
            return this;
        }

        public JdbcInputFormatBuilder setRowTypeInfo(RowTypeInfo rowTypeInfo) {
            format.rowTypeInfo = rowTypeInfo;
            return this;
        }

        public JdbcInputFormatBuilder setFetchSize(int fetchSize) {
            Preconditions.checkArgument(
                    fetchSize == Integer.MIN_VALUE || fetchSize > 0,
                    "Illegal value %s for fetchSize, has to be positive or Integer.MIN_VALUE.",
                    fetchSize);
            format.fetchSize = fetchSize;
            return this;
        }

        public JdbcInputFormatBuilder setAutoCommit(Boolean autoCommit) {
            format.autoCommit = autoCommit;
            return this;
        }

        // 该方法用于完成 JdbcInputFormat对象的构建，并返回该对象。
        // 最后，调用 finish方法返回 JdbcInputFormat对象。
        public JdbcInputFormat finish() {
            
            // 首先使用 JdbcConnectionOptionsBuilder对象构建一个 SimpleJdbcConnectionProvider对象，
            // 并将其设置为 JdbcInputFormat对象的连接提供者。
            format.connectionProvider = new SimpleJdbcConnectionProvider(connOptionsBuilder.build());
            
            // 检查是否提供了行类型信息。如果没有提供行类型信息，
            if (format.queryTemplate == null) {
                // 则抛出 NullPointerException异常，并提示用户没有提供 RowTypeInfo。
                throw new NullPointerException("No query supplied");
            }
            if (format.rowTypeInfo == null) {
                // RowTypeInfo是 Flink中用于描述行数据类型的类，它包含每个字段的名称和类型信息。
                // 在使用 JdbcInputFormat读取 JDBC数据源时，需要指定 RowTypeInfo来解析读取的数据。
                // 因此，如果没有提供 RowTypeInfo，则无法读取数据。
                throw new NullPointerException("No " + RowTypeInfo.class.getSimpleName() + " supplied");
            }
            if (format.parameterValues == null) {
                // 最后，如果没有配置输入分片，则记录一条 DEBUG级别的日志。
                LOG.debug("No input splitting configured (data will be read with parallelism 1).");
            }
            return format;
        }
    }
}
